// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jshin

#pragma once

#include <map>
#include <memory>
#include <set>
#include <string>
#include <vector>

#include "coin2/exchange/base/executor/async_executor.h"
#include "coin2/exchange/base/executor/live_executor.h"
#include "coin2/exchange/base/symbology/product.h"

namespace coin2::exchange::base::api_base {

using coin2::exchange::base::executor::AsyncExecutor;
using coin2::exchange::base::executor::LiveAsyncExecutor;
using coin2::exchange::base::symbology::IProduct;

struct HealthReport {
  std::string report;
  int connection_num;
  int symbol_num;
};

class BaseRawAsyncSubscriber : public IAsyncActor {
 public:
  explicit BaseRawAsyncSubscriber(LiveAsyncExecutor* flow_sub) : flow_sub_(flow_sub) {}

  const std::set<std::string>& native_symbols() const { return native_symbols_; }
  const std::map<char, std::vector<std::string>> & native_symbol_groups() const {
    return native_symbol_groups_;
  }

  void RequestNativeSymbol(const std::string& native_symbol, const char group) {
    CHECK_EQ(native_symbols_.count(native_symbol), 0) << native_symbol;
    native_symbols_.insert(native_symbol);
    native_symbol_groups_[group].push_back(native_symbol);
  }

  void RemoveNativeSymbol(const std::string& native_symbol, const char group) {
    native_symbols_.erase(native_symbol);
    if (native_symbol_groups_.count(group) > 0) {
      auto& symbols = native_symbol_groups_[group];
      auto iter = std::find(symbols.begin(), symbols.end(), native_symbol);
      if (iter != symbols.end()) {
        symbols.erase(iter);
        if (symbols.empty()) {
          native_symbol_groups_.erase(group);
        }
      }
    }
  }

  char FindNativeSymbolGroup(const std::string& native_symbol) const {
    // Not a good implementation, as this function is seldom used.
    for (auto& iter : native_symbol_groups_) {
      if (std::find(iter.second.begin(), iter.second.end(), native_symbol) != iter.second.end()) {
        return iter.first;
      }
    }
    return '\0';
  }

  boost::asio::io_context* ioc() { return flow_sub_->RawIoc(); }

  void StartAsync() override {
    CHECK(!sub_running_);
    sub_running_ = true;

    SubscribeAsync();
  }

  void StopAsync() override {
    CHECK(sub_running_);
    sub_running_ = false;
    CleanUp();
  }

  virtual void CleanUp() {}

  virtual void GetHealthReport(int indent, HealthReport* health) const {}
  virtual void SubscribeAsync() = 0;
  virtual void RegisterPubTopic() = 0;
  virtual void RegisterPubTopic(const std::string& topic) {
     flow_sub_->distributor()->Publish(topic);
  };

  virtual bool BanProduct(const std::vector<const IProduct*>& products) { return false; }

 protected:
  std::set<std::string> native_symbols_;
  std::map<char, std::vector<std::string>> native_symbol_groups_;

  LiveAsyncExecutor* flow_sub_;
  volatile bool sub_running_ {false};
};

}  // namespace coin2::exchange::base::api_base
